package com.ficus.road.maintain.grpc;

import com.ficus.road.maintain.core.model.oss.AliyunOssResult;
import com.ficus.road.maintain.service.DataProcessorManager;
import com.ficus.road.maintain.util.AliyunOSSUtil;
import com.google.protobuf.ByteString;
import ficus.car_demo.FrameMeta;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.sql.Timestamp;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.zip.GZIPOutputStream;

import static com.ficus.road.maintain.util.SsfFileUtil.*;

@Slf4j
public class FrameMetaProcessor implements Runnable {

    public static final int INTERVAL = 4000;
    public BlockingDeque<List<FrameMeta.FrameResult>> blockingDeque;
    public DateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    String prefix;
    Map<String, Integer> map;
    AliyunOSSUtil aliyunOSSUtil;
    DataProcessorManager dataProcessorManager;
    private volatile Long lastInspectTime = 0l;

    public FrameMetaProcessor(String prefix, AliyunOSSUtil aliyunOSSUtil, DataProcessorManager dataProcessorManager) {
        this.prefix = prefix;
        this.aliyunOSSUtil = aliyunOSSUtil;
        this.dataProcessorManager = dataProcessorManager;
        this.blockingDeque = new LinkedBlockingDeque<>(50);
        this.map = new ConcurrentHashMap<>();
    }

    public static byte[] compress(byte[] bytes) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        GZIPOutputStream gzip;
        try {
            gzip = new GZIPOutputStream(out);
            gzip.write(bytes);
            gzip.close();
        } catch (IOException e) {
            log.error("gzip compress error.", e);
        }
        return out.toByteArray();
    }

    @Override
    public void run() {
        while (true) {
            try {
                //把uuid封装了一下
                List<FrameMeta.FrameResult> frameResultList = blockingDeque.take();
                onProcess(frameResultList);
            } catch (Exception e) {
                log.error("Base message processor error", e);
            }
        }
    }

    /**
     * 普通插入，知道有空间再插入
     *
     * @param frameResultList
     * @throws InterruptedException
     */
    public void put(List<FrameMeta.FrameResult> frameResultList) throws InterruptedException {
        blockingDeque.put(frameResultList);
    }

    public void onProcess(List<FrameMeta.FrameResult> list) {
        int size = list.size();
        long timeStamp = list.get(0).getTimeStamp();
        String carId = list.get(0).getCarId();
        String formatDate = sdf.format(new Timestamp(timeStamp));
        sdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        String directory = generateDirectory(carId, formatDate);
        String inspectDirectory = directory.replace(PIC, INSPECT);
        String fileName = directory + "/" + timeStamp;

        FrameMeta.PicArray.Builder picArrayBuilder = FrameMeta.PicArray.newBuilder();
        FrameMeta.PicArray.Builder diseasePicArrayBuilder = FrameMeta.PicArray.newBuilder();
        List<FrameMeta.FrameResult> newList = new ArrayList<>();
        int i = 0;
        int j = 0;
        List<FrameMeta.InspectPic> inspectPics = new ArrayList<>();
        for (FrameMeta.FrameResult frame : list) {
            FrameMeta.FrameResult.Builder builder = frame.toBuilder();
            if (!frame.getPicture().isEmpty()) {
                picArrayBuilder.addPics(frame.getPicture());
                builder.setPicture(ByteString.copyFromUtf8("" + i++));
            }
            if (!frame.getLidarResult().getPicture().isEmpty()) {
                picArrayBuilder.addPics(frame.getLidarResult().getPicture());
                builder.setLidarResult(builder.getLidarResultBuilder().setPicture(ByteString.copyFromUtf8("" + i++)).build());
            }
            if (!frame.getPerceptionResultSet().getSegmentPicture().isEmpty()) {
                picArrayBuilder.addPics(frame.getPerceptionResultSet().getSegmentPicture());
                builder.setPerceptionResultSet(builder.getPerceptionResultSetBuilder().setSegmentPicture(ByteString.copyFromUtf8("" + i++)).build());
            }
            if (!frame.getPictureForDisease().isEmpty()) {
                diseasePicArrayBuilder.addPics(frame.getPictureForDisease());
                builder.setPictureForDisease(ByteString.copyFromUtf8("" + j++));
                if (frame.getTimeStamp() - lastInspectTime > INTERVAL) {
                    String inspectFileName = inspectDirectory + "/" + frame.getTimeStamp();
                    AliyunOssResult ossResult = aliyunOSSUtil.uploadImg(inspectFileName, frame.getPictureForDisease().toByteArray());
                    FrameMeta.InspectPic inspectPic = FrameMeta.InspectPic.newBuilder().setTimestamp(frame.getTimeStamp()).setUrl(ossResult.getUrl()).build();
                    inspectPics.add(inspectPic);
                    lastInspectTime = frame.getTimeStamp();
                }
            }
            newList.add(builder.build());
        }
        byte[] compress = compress(picArrayBuilder.build().toByteArray());
        aliyunOSSUtil.uploadData(fileName, compress);
        String diseaseFileName = fileName.replace(PIC, DISEASE);
        aliyunOSSUtil.uploadData(diseaseFileName, diseasePicArrayBuilder.build().toByteArray());
        FrameMeta.SendMetaRequest request = FrameMeta.SendMetaRequest.newBuilder().addAllFrames(newList)
                .setCarId(carId).setEndTime(list.get(size - 1).getTimeStamp()).setPicturePath(fileName)
                .setStartTime(list.get(0).getTimeStamp()).addAllInspectPics(inspectPics).setIsCompressed(true).build();
        dataProcessorManager.put(request);
    }

    private String generateDirectory(String carId, String formatDate) {
        return prefix + "/" + carId + "/" + formatDate + "/" + PIC;
    }
}
